package com.pactera.rabbitmq.utils;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Joiner;
import com.pactera.rabbitmq.bean.MessageInfo;
import com.pactera.rabbitmq.constants.ExchageType;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.Properties;

/**
 * @author: liqingdian
 **/
public class RabbitUtils {

    private RabbitUtils() {
    }

    private static AmqpAdmin amqpAdmin;

    public static RabbitUtils builder(AmqpAdmin amqpAdmin) {
        RabbitUtils.amqpAdmin = amqpAdmin;
        return new RabbitUtils();
    }

    public static AmqpAdmin amqpAdmin() {
        return amqpAdmin;
    }

    public static RabbitTemplate rabbitTemplate() {
        RabbitAdmin rabbitAdmin = (RabbitAdmin) RabbitUtils.amqpAdmin;
        return rabbitAdmin.getRabbitTemplate();
    }

    /**
     * 实时数据同步-发送
     *
     * @param message 实时数据
     * @author: liqingdian
     */
    public static void sendRtds(MessageInfo message) {
        Assert.isNotNull(message, "[message]不能为空");
        Assert.isNotBlank(message.getQueue(), "[data.queue]不能为空");
        Assert.isNotBlank(message.getTag(), "[data.tag]不能为空");
        Assert.isNotBlank(message.getData(), "[data.data]不能为空");
        MessageProperties properties = new MessageProperties();
        properties.setConsumerQueue(message.getQueue());
        properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        properties.setPriority(0);
        Date timestamp = new Date();
        properties.setTimestamp(timestamp);
        properties.setType(message.getTag());
        properties.setMessageId(Joiner.on("-").join(message.getQueue(), message.getTag(), timestamp.getTime()));
        Long expiration = message.getExpiration();
        if (expiration != null && expiration > 0) {
            properties.setExpiration(String.valueOf(expiration));
        }
        String msg = JSON.toJSONString(message);
        Message sendMsg = new Message(msg.getBytes(StandardCharsets.UTF_8), properties);
        RabbitUtils.rabbitTemplate().convertAndSend(AdminUtils.EXCHANGE_RTDS, message.getQueue(), sendMsg);
    }

    /**
     * 声明交互机
     *
     * @param name       交互机名称
     * @param type       类型:direct-简单模式(默认),topic-主题模式，fanout-广播模式，headers-键值对模式
     * @param durable    是否持久化
     * @param autoDelete 是否自动删除
     * @param arguments  结构化参数
     * @author: liqingdian
     */
    public static void exchangeDeclare(String name, ExchageType type, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 声明交换机
        //direct-普通,topic-主题,headers-键值对,fanout-广播
        Exchange exchange;
        switch (type) {
            case TOPIC:
                exchange = new TopicExchange(name, durable, autoDelete, arguments);
                break;
            case FANOUT:
                exchange = new FanoutExchange(name, durable, autoDelete, arguments);
                break;
            case HEADERS:
                exchange = new HeadersExchange(name, durable, autoDelete, arguments);
                break;
            default:
                exchange = new DirectExchange(name, durable, autoDelete, arguments);
        }
        amqpAdmin.declareExchange(exchange);
    }

    /**
     * 删除交互机
     *
     * @param name 交互机名称
     * @author: liqingdian
     */
    public static void exchangeDelete(String name) {
        amqpAdmin.deleteExchange(name);
    }

    /**
     * 声明队列
     *
     * @param name       队列名称
     * @param durable    是否持久化
     * @param exclusive  是否排他
     * @param autoDelete 是否自动删除
     * @param arguments  结构化参数
     * @author: liqingdian
     */
    public static void queueDeclare(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        // 声明队列
        amqpAdmin.declareQueue(new Queue(name, durable, exclusive, autoDelete, arguments));
    }

    /**
     * 删除队列
     *
     * @param name 队列名称
     * @author: liqingdian
     */
    public static void queueDelete(String name) {
        amqpAdmin.deleteQueue(name);
    }

    /**
     * 清空队列
     *
     * @param name 队列名称
     * @author: liqingdian
     */
    public static void purgeQueue(String name) {
        amqpAdmin.purgeQueue(name, false);
    }

    /**
     * 获取队列配置
     *
     * @param name 队列名称
     * @author: liqingdian
     */
    public static Properties getQueueProperties(String name) {
        return amqpAdmin.getQueueProperties(name);
    }


    /**
     * 删除队列
     *
     * @param name     队列名称
     * @param ifUnused true:队列未使用才删除
     * @param ifEmpt   true:队列为空才删除
     * @author: liqingdian
     */
    public static void queueDelete(String name, boolean ifUnused, boolean ifEmpt) {
        amqpAdmin.deleteQueue(name, ifUnused, ifEmpt);
    }

    /**
     * 绑定交互机
     *
     * @param binding      绑定队列/交互机名称
     * @param exchangeBind 是否交互机绑定
     * @param exchange     绑定到该交互机
     * @param routingKey   绑定路由Key，为空则以binding作为key
     * @param arguments    结构化参数
     * @author: liqingdian
     */
    public static void bind(String binding, boolean exchangeBind, String exchange, String routingKey, Map<String, Object> arguments) {
        // 绑定
        if (StringUtils.isBlank(routingKey)) {
            routingKey = binding;
        }
        Binding.DestinationType type;
        if (exchangeBind) {
            type = Binding.DestinationType.EXCHANGE;
        } else {
            type = Binding.DestinationType.QUEUE;
        }
        amqpAdmin.declareBinding(new Binding(binding, type, exchange, routingKey, arguments));
    }

    /**
     * 解绑交互机
     *
     * @param binding      绑定队列/交互机名称
     * @param exchangeBind 是否交互机绑定
     * @param exchange     绑定到该交互机
     * @param routingKey   绑定路由Key
     * @param arguments    结构化参数
     * @author: liqingdian
     */
    public static void unbind(String binding, boolean exchangeBind, String exchange, String routingKey, Map<String, Object> arguments) {
        Binding.DestinationType type;
        if (exchangeBind) {
            type = Binding.DestinationType.EXCHANGE;
        } else {
            type = Binding.DestinationType.QUEUE;
        }
        Binding bind = new Binding(binding, type, exchange, routingKey, arguments);
        amqpAdmin.removeBinding(bind);
    }
}
